package com.central.dataManage.common.datasourcePlugin;

import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLCommentHint;
import com.alibaba.druid.sql.ast.SQLDataTypeImpl;
import com.alibaba.druid.sql.ast.SQLObject;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.SQLCharExpr;
import com.alibaba.druid.sql.ast.statement.SQLAlterTableStatement;
import com.alibaba.druid.sql.ast.statement.SQLColumnDefinition;
import com.alibaba.druid.sql.ast.statement.SQLCreateTableStatement;
import com.alibaba.druid.sql.ast.statement.SQLTableElement;
import com.alibaba.druid.sql.dialect.hive.stmt.HiveCreateTableStatement;
import com.alibaba.druid.sql.parser.ParserException;
import com.alibaba.druid.sql.parser.SQLParserUtils;
import com.alibaba.druid.sql.parser.SQLStatementParser;
import com.alibaba.druid.sql.visitor.SQLASTVisitor;
import com.alibaba.druid.sql.visitor.VisitorFeature;
import com.alibaba.druid.util.JdbcConstants;
import com.central.common.constant.CommonConstant;
import com.central.common.enums.DbType;
import com.central.common.model.ColumnInfo;
import com.central.common.model.IndexInfo;
import com.central.dataManage.common.utils.PropertyUtils;
import com.central.dataManage.model.DatabaseInfo;
import com.central.common.model.DatasourceInfo;
import com.central.common.model.TableInfo;
import io.itit.itf.okhttp.FastHttpClient;
import io.itit.itf.okhttp.util.FileUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.sql.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import static com.central.dataManage.common.Constants.*;

/**
 * @author Tindy
 * @date 2021/7/5
 * @describe hive数据源插件
 */
public class HiveDatasourcePlugin extends BaseDatasourcePlugin {
    public HiveDatasourcePlugin(DatasourceInfo datasource) {
        super(datasource);
    }

    @Override
    public String getJdbcUrl() {
        String address = "jdbc:hive2://" + getDatasource().getHost() + ":" + getDatasource().getPort() + "/";
        String jdbcUrl = address;
        if (!StringUtils.isBlank(getDatasource().getDbName())) {
            jdbcUrl = address + getDatasource().getDbName();
        }
        String other = buildOther(getDatasource().getOther(), ";");
        if (StringUtils.isNotEmpty(other)) {
            jdbcUrl += ";" + other;
        }
        return jdbcUrl;
    }

    @Override
    public Connection getConn() throws SQLException {
        Connection conn = null;
        try {
            String driver=StringUtils.isNotEmpty(getDatasource().getDriver())?getDatasource().getDriver():getDatasourceDriver();
            Class.forName(driver);
            if(StringUtils.isNotBlank(getDatasource().getKeytabUrl()) && StringUtils.isNotBlank(getDatasource().getKrb5ConfUrl())){
                synchronized (CommonConstant.KRB5CONF_LOCK) {
                    UserGroupInformation.setLoginUser(null);
                    System.clearProperty("java.security.krb5.conf");
                    String krb5ConfPath=getKrb5ConfPath(getDatasource());
                    System.setProperty("java.security.krb5.conf", krb5ConfPath);
                    Configuration conf = new Configuration();
                    conf.set("hadoop.security.authentication", "Kerberos");
                    UserGroupInformation.setConfiguration(conf);
                    String keytabPath = getKeyTabPath(getDatasource());
                    UserGroupInformation.loginUserFromKeytab(getDatasource().getUsername(), keytabPath);
                    conn=DriverManager.getConnection(getJdbcUrl());
                }
            }else {
                conn = DriverManager.getConnection(getJdbcUrl());
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return conn;
    }

    private String getKrb5ConfPath(DatasourceInfo datasourceInfo) throws Exception {
        File file = new File(PropertyUtils.getString(STORE_KEYTAB_DIR,STORE_KEYTAB_DIR_DEFAULT) + File.separator + datasourceInfo.findKrb5ConfName());
        if(!file.exists()){
            InputStream is= FastHttpClient.get().url(datasourceInfo.getKrb5ConfUrl()).build().execute().byteStream();
            FileUtil.saveContent(is, file);
            is.close();
        }
        return file.getPath();
    }


    private String getKeyTabPath(DatasourceInfo datasourceInfo) throws Exception {
        File file = new File(PropertyUtils.getString(STORE_KEYTAB_DIR,STORE_KEYTAB_DIR_DEFAULT) + File.separator + datasourceInfo.findKeytabName());
        if(!file.exists()){
            InputStream is= FastHttpClient.get().url(datasourceInfo.getKeytabUrl()).build().execute().byteStream();
            FileUtil.saveContent(is, file);
            is.close();
        }
        return file.getPath();
    }

    @Override
    public List<DatabaseInfo> getDatabases(Connection conn) throws SQLException {

        ArrayList<DatabaseInfo> databases = new ArrayList<>();
        if(!StringUtils.isBlank(getDatasource().getDbName())){
            DatabaseInfo database = new DatabaseInfo();
            database.setDbName(getDatasource().getDbName());
            database.setDatasourceId(getDatasource().getId());
            database.setDatasourceType(DbType.HIVE);
            databases.add(database);
        }else{
            Statement statement = conn.createStatement();
            ResultSet resultSet = statement.executeQuery("SHOW DATABASES");
            while (resultSet.next()) {
                String dbName = resultSet.getString(1);
                DatabaseInfo database = new DatabaseInfo();
                database.setDbName(dbName);
                database.setDatasourceId(getDatasource().getId());
                database.setDatasourceType(DbType.HIVE);
                databases.add(database);
            }
            resultSet.close();
            statement.close();
        }
        return databases;
    }

    @Override
    public List<TableInfo> getTables(DatabaseInfo database, Connection conn) throws SQLException {
        ArrayList<TableInfo> tables = new ArrayList<>();
        Statement statement = conn.createStatement();
        ResultSet resultSet = statement.executeQuery("SHOW TABLES FROM `" + database.getDbName() + "`");
        while (resultSet.next()) {
            String tableName = resultSet.getString(1);
            TableInfo table = new TableInfo();
            table.setDatasourceId(getDatasource().getId());
            table.setDatabaseId(database.getId());
            table.setDbName(database.getDbName());
            table.setTableName(tableName);
            table.setDatasourceType(DbType.HIVE);
            tables.add(table);
        }
        resultSet.close();
        statement.close();
        return tables;
    }

    @Override
    public List<ColumnInfo> getColumns(DatabaseInfo database, TableInfo table, Connection conn) throws SQLException {
        List<ColumnInfo> columns = new ArrayList<>();
        Statement statement = conn.createStatement();
        ResultSet resultSet = statement.executeQuery("SHOW CREATE TABLE `" + database.getDbName() + "`." + table.getTableName());
        StringBuilder createTableDdl = new StringBuilder();
        StringBuilder fullDdl = new StringBuilder();
        while (resultSet.next()) {
            String s = resultSet.getString(1);
            fullDdl.append(s).append("\r");
            if ("ROW FORMAT SERDE ".equals(s)) break;
            createTableDdl.append(s);
        }
        while (resultSet.next()) {
            String s = resultSet.getString(1);
            fullDdl.append(s).append("\r");;
        }
        table.setDdl(fullDdl.toString());
        table.setTableComment(HiveDdlParse.getTableComment(createTableDdl.toString()));
        columns=HiveDdlParse.getColumnInfo(createTableDdl.toString());
        resultSet.close();
        statement.close();
        return columns;
    }

    @Override
    public List<IndexInfo> getIndexs(DatabaseInfo database, TableInfo table, Connection conn) throws SQLException {
        return new ArrayList<IndexInfo>();
    }

    @Override
    public void createNewDatabase(String dbName) throws SQLException {
        Statement statement = getConn().createStatement();
        statement.execute("create database " + dbName);
        statement.close();
    }

    @Override
    public void dropDatabase(String dbName) throws SQLException {
        Statement statement = getConn().createStatement();
        statement.execute("drop database " + dbName);
        statement.close();
    }

    @Override
    public void dropTable(String dbName, String tableName) throws SQLException {
        Statement statement = getConn().createStatement();
        statement.execute("drop table " + dbName + "." + tableName);
        statement.close();
    }

    @Override
    public List<SQLStatement> checkCreateTableDdl(String ddl) throws SQLException {
        SQLStatementParser parser = SQLParserUtils.createSQLStatementParser(ddl, JdbcConstants.HIVE);
        List<SQLStatement> sqlStatements = parser.parseStatementList();
        if(sqlStatements.size()>1){
            throw new ParserException("只允许一条建表语句");
        }
        if(!(sqlStatements.get(0) instanceof SQLCreateTableStatement)){
            throw new ParserException("sql必须是建表语句");
        }
        return sqlStatements;
    }

    @Override
    public void createTableByDdl(DatabaseInfo database, List<SQLStatement> sqlStatements) throws SQLException {
        Connection conn = getConn();
        try {
            Statement statement = conn.createStatement();
            statement.execute("use `"+database.getDbName()+"`");
            for (SQLStatement sqlStatement : sqlStatements) {
                statement.execute(sqlStatement.toString());
            }
        }finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Override
    public String generateDdlSql(TableInfo table) {
        HiveCreateTableStatement statement = new HiveCreateTableStatement();
        statement.setName(table.getTableName());
        statement.setComment(new SQLCharExpr(table.getTableComment()));
        List<SQLColumnDefinition> partitionColumns = statement.getPartitionColumns();
        List<SQLTableElement> tableElementList = statement.getTableElementList();
        List<ColumnInfo> columns = table.getColumns();
        for (ColumnInfo column : columns) {
            SQLColumnDefinition element = new SQLColumnDefinition();
            element.setName(column.getColumnName());
            SQLDataTypeImpl sqlDataType = new SQLDataTypeImpl(column.getDataType());
            element.setDataType(sqlDataType);
            if(column.getColumnComment()!=null){
                element.setComment(column.getColumnComment());
            }
            if(column.getIsPartition()){
                partitionColumns.add(element);
            }else {
                tableElementList.add(element);
            }

        }
        return statement.toString();
    }

    @Override
    public String getTableNameByDdl(String ddl) {
        SQLStatementParser parser = SQLParserUtils.createSQLStatementParser(ddl, JdbcConstants.HIVE);
        SQLCreateTableStatement sqlCreateTableStatement = parser.getSQLCreateTableParser().parseCreateTable();
        String tableName = sqlCreateTableStatement.getName().getSimpleName();
        if (tableName.matches("`.*`")) {
            tableName = tableName.substring(1, tableName.length() - 1);
        }
        return tableName;
    }

    @Override
    public String getTableNameByModifyDdl(SQLStatement sqlStatement) {
        SQLAlterTableStatement sqlAlterTableStatement = (SQLAlterTableStatement) sqlStatement;
        String tableName =sqlAlterTableStatement.getTableName();
        if (tableName.matches("`.*`")) {
            tableName = tableName.substring(1, tableName.length() - 1);
        }
        return tableName;
    }

    @Override
    public String generateModifySql(TableInfo table, TableInfo oldTable) throws Exception {
        StringBuilder sqls=new StringBuilder();
        List<ColumnInfo> columns = table.getColumns();
        List<ColumnInfo> oldColumns = oldTable.getColumns();
        int i=0;
        //修改列
        for (;i<oldColumns.size();i++){
            ColumnInfo column = columns.get(i);
            if(!column.getColumnName().equals(oldColumns.get(i).getColumnName())){
                throw new Exception("修改列字段与原表不匹配");
            }
            String sql="ALTER TABLE  "+table.getTableName()+" CHANGE "+column.getColumnName()+
                    " "+column.getColumnName()+" "+column.getDataType();
            if(column.getColumnComment()!=null){
                sql+=" COMMENT \""+column.getColumnComment()+"\"";
            }
            sqls.append(sql).append(";");
        }
        //新增列
        for(;i<columns.size();i++){
            ColumnInfo column = columns.get(i);
            String addSql="ALTER TABLE  "+table.getTableName()+" ADD COLUMNS ("+
                    column.getColumnName()+" "+column.getDataType();;
            if(column.getColumnComment()!=null){
                addSql+=" COMMENT \""+column.getColumnComment()+"\")";
            }
            sqls.append(addSql).append(";");
        }
        return sqls.toString();
    }

    @Override
    public void modifyTableByDdl(DatabaseInfo database, List<SQLStatement> sqlStatements) throws SQLException {
        createTableByDdl(database,sqlStatements);
    }

    @Override
    public List<SQLStatement> checkModifyTableDdl(String ddl,String tableName) {
        SQLStatementParser parser = SQLParserUtils.createSQLStatementParser(ddl, JdbcConstants.HIVE);
        List<SQLStatement> sqlStatements = parser.parseStatementList();
        for (SQLStatement sqlStatement : sqlStatements) {
            if (!(sqlStatement instanceof SQLAlterTableStatement)) {
                throw new ParserException("修改语句类型错误");
            }
            if(!tableName.equals(getTableNameByModifyDdl(sqlStatement))){
                throw new ParserException("修改语句表名与原表不匹配");
            }
        }
        return sqlStatements;
    }

    public static void main(String[] args) {
        String ddl="select * from aa;truncate table add;select ${a;d} from bv;create table ${aa}(string a);show databases;insert into aa select cd.df,cd.aaa,ab.${sd} from cd join ab on cd.id=ab.id";
        SQLStatementParser parser = SQLParserUtils.createSQLStatementParser(ddl, JdbcConstants.HIVE);
        List<SQLStatement> sqlStatements = parser.parseStatementList();
        System.out.println(sqlStatements.get(0).toLowerCaseString()
);
    }

    @Override
    public List<String> getDataTypes() {
        String[] dataTypes={"STRING","BOOLEAN","TIMESTAMP","INT","DATE","TINYINT","SMALLINT","BIGINT","FLOAT","DOUBLE","DEICIMAL","VARCHAR","CHAR","BINARY","ARRAY","MAP"};
        return Arrays.asList(dataTypes);
    }

    @Override
    public String format(String sql) {
        return SQLUtils.format(sql, JdbcConstants.HIVE);
    }

    @Override
    public String getDatasourceDriver() {
        return ORG_APACHE_HIVE_JDBC_HIVE_DRIVER;
    }

    private class HiveSqlStatement implements SQLStatement{

        private String sqls;

        public String getSqls() {
            return sqls;
        }

        public void setSqls(String sqls) {
            this.sqls = sqls;
        }

        @Override
        public com.alibaba.druid.DbType getDbType() {
            return null;
        }

        @Override
        public boolean isAfterSemi() {
            return false;
        }

        @Override
        public void setAfterSemi(boolean b) {

        }

        @Override
        public void accept(SQLASTVisitor sqlastVisitor) {

        }

        @Override
        public SQLStatement clone() {
            return null;
        }

        @Override
        public SQLObject getParent() {
            return null;
        }

        @Override
        public void setParent(SQLObject sqlObject) {

        }

        @Override
        public Map<String, Object> getAttributes() {
            return null;
        }

        @Override
        public boolean containsAttribute(String s) {
            return false;
        }

        @Override
        public Object getAttribute(String s) {
            return null;
        }

        @Override
        public void putAttribute(String s, Object o) {

        }

        @Override
        public Map<String, Object> getAttributesDirect() {
            return null;
        }

        @Override
        public void output(StringBuffer stringBuffer) {

        }

        @Override
        public void output(Appendable appendable) {

        }

        @Override
        public void addBeforeComment(String s) {

        }

        @Override
        public void addBeforeComment(List<String> list) {

        }

        @Override
        public List<String> getBeforeCommentsDirect() {
            return null;
        }

        @Override
        public void addAfterComment(String s) {

        }

        @Override
        public void addAfterComment(List<String> list) {

        }

        @Override
        public List<String> getAfterCommentsDirect() {
            return null;
        }

        @Override
        public boolean hasBeforeComment() {
            return false;
        }

        @Override
        public boolean hasAfterComment() {
            return false;
        }

        @Override
        public List<SQLObject> getChildren() {
            return null;
        }

        @Override
        public String toLowerCaseString() {
            return null;
        }

        @Override
        public String toParameterizedString() {
            return null;
        }

        @Override
        public List<SQLCommentHint> getHeadHintsDirect() {
            return null;
        }

        @Override
        public void setHeadHints(List<SQLCommentHint> list) {

        }

        @Override
        public String toString() {
            return sqls;
        }

        @Override
        public String toString(VisitorFeature... visitorFeatures) {
            return null;
        }
    }
}
